package com.galo.service.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.galo.service.config.MqttProperties;
import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.AmqpTemplate;
import redis.clients.jedis.Jedis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.util.Random;

/**
 * 消息接收处理类
 */
@Configuration
@Slf4j
public class MqttInboundConfiguration {
	private static Jedis jedis=new Jedis("localhost",6379,100000);

	@Autowired
	private AmqpTemplate rabbitTemplate;
    @Autowired
    private MqttProperties mqttProperties;
    private Random rand = new Random();
    private String clientId = "";
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttPahoClientFactory) {
        String[] inboundTopics = mqttProperties.getInbound().getTopics().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInbound().getUrl(), clientId + String.format("%05d", rand.nextInt(1000000000)),
                        mqttPahoClientFactory,inboundTopics);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {

                log.info("MQTT收到消息："+(String) message.getPayload());

                String submsg = (String) message.getPayload();
                JSONObject object = JSONObject.parseObject(submsg);
                String action = object.getString("action");
                if(action.equals("rep") ) {
                	//直接将接收到的数据存到队列中去
                    rabbitTemplate.convertAndSend("topic.rep", (String) message.getPayload());
                }else if(action.equals("auto")) {
                	//直接将接收到的数据存到队列中去
                    rabbitTemplate.convertAndSend("topic.auto", (String) message.getPayload());
                }
                
            }

        };
    }
}
